package com.yu.rpc.core.proxy;

import cn.hutool.core.collection.CollUtil;
import cn.hutool.core.util.IdUtil;
import cn.hutool.http.HttpRequest;
import cn.hutool.http.HttpResponse;
import com.yu.rpc.core.RpcApplication;
import com.yu.rpc.core.config.RpcConfig;
import com.yu.rpc.core.constant.RpcConstant;
import com.yu.rpc.core.fault.retry.RetryStrategy;
import com.yu.rpc.core.fault.retry.RetryStrategyFactory;
import com.yu.rpc.core.fault.tolerant.TolerantStrategy;
import com.yu.rpc.core.fault.tolerant.TolerantStrategyFactory;
import com.yu.rpc.core.loadbalancer.LoadBalancerFactory;
import com.yu.rpc.core.model.RPCRequest;
import com.yu.rpc.core.model.RPCResponse;
import com.yu.rpc.core.model.ServiceMetaInfo;
import com.yu.rpc.core.protocol.ProtocolMessage;
import com.yu.rpc.core.protocol.ProtocolMessageDecoder;
import com.yu.rpc.core.protocol.ProtocolMessageEncoder;
import com.yu.rpc.core.protocol.ProtocolMessageTypeEnum;
import com.yu.rpc.core.registry.Registry;
import com.yu.rpc.core.registry.RegistryFactory;
import com.yu.rpc.core.serializer.Serializer;
import com.yu.rpc.core.serializer.SerializerFactory;
import com.yu.rpc.core.server.tcp.TcpBufferHandlerWrapper;
import io.vertx.core.Vertx;
import io.vertx.core.buffer.Buffer;
import io.vertx.core.net.NetClient;
import io.vertx.core.net.NetSocket;

import java.io.IOException;
import java.lang.reflect.InvocationHandler;
import java.lang.reflect.Method;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;

/**
 * 服务代理（JDK 动态代理）
 */
public class ServiceProxy implements InvocationHandler {

    /**
     * 调用代理
     */
    @Override
    public Object invoke(Object proxy, Method method, Object[] args) throws Throwable {
        String serviceName = method.getDeclaringClass().getName();
        RpcConfig rpcConfig = RpcApplication.getConfig();
        try {
            // 从注册中心获取服务提供者请求地址
            Registry registry = RegistryFactory.getInstance(rpcConfig.getRegistryConfig().getRegistry());
            ServiceMetaInfo serviceMetaInfo = new ServiceMetaInfo();
            serviceMetaInfo.setServiceName(serviceName);
            serviceMetaInfo.setServiceVersion(RpcConstant.DEFAULT_SERVICE_VERSION);
            List<ServiceMetaInfo> serviceMetaInfoList = registry.serviceDiscovery(serviceMetaInfo.getServiceKey());
            if (CollUtil.isEmpty(serviceMetaInfoList)) {
                throw new RuntimeException("暂无服务地址");
            }
            // 构造请求
            RPCRequest rpcRequest = RPCRequest.builder()
                    .serviceName(serviceName)
                    .methodName(method.getName())
                    .paramTypes(method.getParameterTypes())
                    .params(args)
                    .build();
            //负载均衡
            Map<String, Object> requestParams = new HashMap<>();
            requestParams.put("methodName", rpcRequest.getMethodName());
            ServiceMetaInfo selectedServiceMetaInfo = LoadBalancerFactory.getInstance(rpcConfig.getLoadBalancer()).select(requestParams, serviceMetaInfoList);
            //重试机制
            RetryStrategy retryStrategy = RetryStrategyFactory.getInstance(rpcConfig.getRetryStrategy());
            RPCResponse rpcResponse = retryStrategy.doRetry(() -> {
                // 发送请求
                return doTcpRequest(selectedServiceMetaInfo, rpcRequest);
            });
            return rpcResponse.getResult();
        } catch (Exception e) {
            TolerantStrategy tolerantStrategy = TolerantStrategyFactory.getInstance(rpcConfig.getTolerantStrategy());
            tolerantStrategy.doTolerant(null, e);
        }
        return null;
    }

    private RPCResponse doHttpRequest(ServiceMetaInfo metaInfo, RPCRequest rpcRequest) throws IOException {
        // 指定序列化器
        Serializer serializer = SerializerFactory.getSerializer(RpcApplication.getConfig().getSerializer());
        // 序列化
        byte[] bodyBytes = serializer.serialize(rpcRequest);
        try (HttpResponse httpResponse = HttpRequest.post(metaInfo.getServiceAddress())
                .body(bodyBytes)
                .execute()) {
            byte[] result = httpResponse.bodyBytes();
            return serializer.deserialize(result, RPCResponse.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
    }

    private RPCResponse doTcpRequest(ServiceMetaInfo metaInfo, RPCRequest rpcRequest) throws ExecutionException, InterruptedException {
        Vertx vertx = Vertx.vertx();
        NetClient netClient = vertx.createNetClient();
        CompletableFuture<RPCResponse> future = new CompletableFuture<>();
        netClient.connect(metaInfo.getServicePort(), metaInfo.getServiceHost(), result -> {
            if (result.succeeded()) {
                NetSocket socket = result.result();
                ProtocolMessage<RPCRequest> protocolMessage = new ProtocolMessage<>();
                ProtocolMessage.Header header = ProtocolMessage.Header.defaultHeader();
                header.setType((byte) ProtocolMessageTypeEnum.REQUEST.getKey());
                header.setRequestId(IdUtil.getSnowflakeNextId());
                protocolMessage.setHeader(header);
                protocolMessage.setBody(rpcRequest);
                try {
                    Buffer encoderBuffer = ProtocolMessageEncoder.encode(protocolMessage);
                    socket.write(encoderBuffer);
                } catch (IOException e) {
                    e.printStackTrace();
                }
                TcpBufferHandlerWrapper handlerWrapper = new TcpBufferHandlerWrapper(buffer -> {
                    try {
                        ProtocolMessage<RPCResponse> response = (ProtocolMessage<RPCResponse>) ProtocolMessageDecoder.decode(buffer);
                        future.complete(response.getBody());
                    } catch (IOException e) {
                        throw new RuntimeException(e);
                    }
                });
                socket.handler(handlerWrapper);
            }
        });
        RPCResponse rpcResponse = future.get();
        netClient.close();
        return rpcResponse;
    }
}

